package main

import (
	"fmt"
	"time"

	"github.com/Shopify/sarama"
	"os"
	"github.com/json-iterator/go"
)

// kafka consumer

type Student struct{
	Id string
	Name string
	Birth string
	Sex string
	CreateTime time.Time
	UpdateTime time.Time
	CourseList []*Course
}

type Course struct{
	Id string
	Name string
	TeacherId string
	CreateTime time.Time
	UpdateTime time.Time
	Score Score
	Teacher Teacher
}

type Teacher struct{
	Id string
	Name string
	CreateTime time.Time
	UpdateTime time.Time
}

type Score struct{
	StudentId string
	CourseId  string
	Score string
	CreateTime time.Time
	UpdateTime time.Time
}

func main() {
	consumer, err := sarama.NewConsumer([]string{"127.0.0.1:9092"}, nil)
	if err != nil {
		fmt.Printf("fail to start consumer, err:%v\n", err)
		return
	}
	// partitionList, err := consumer.Partitions("nginx_log") // 根据topic取到所有的分区
	// if err != nil {
	//   fmt.Printf("fail to get list of partition:err%v\n", err)
	//   return
	// }
	// fmt.Println("partitionList：", partitionList)
	// for partition := range partitionList { // 遍历所有的分区
	//   // 针对每个分区创建一个对应的分区消费者
	//   pc, err := consumer.ConsumePartition("web_log", int32(partition), sarama.OffsetNewest)
	//   if err != nil {
	//     fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
	//     return
	//   }
	//   defer pc.AsyncClose()
	//   // 异步从每个分区消费信息
	//   go func(sarama.PartitionConsumer) {
	//     for msg := range pc.Messages() {
	//       fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, msg.Value)
	//     }
	//   }(pc)
	// }

	cp, err := consumer.ConsumePartition("nginx_log", 0, sarama.OffsetOldest)
	if err != nil {
		fmt.Fprintf(os.Stdout, "try create partition_consumer err, message=%v \n", err)
		return
	}
	defer cp.Close()

	for {
		select {
		case msg := <-cp.Messages():
			fmt.Fprintf(os.Stdout, "msg offset: %d, partition: %d, timestamp: %s, value: %s \n", msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
			var s []*Student
			jerr := jsoniter.ConfigCompatibleWithStandardLibrary.Unmarshal(msg.Value, &s)
			if jerr != nil{
				fmt.Println("jerr：", jerr)
			}
			fmt.Println("s：", s)
		case err := <-cp.Errors():
			fmt.Fprintf(os.Stdout, "err :%v \n", err)
		}
	}
}
